package com.yc.bigdata.hbase.core;

import com.google.common.collect.Maps;
import com.yc.bigdata.hbase.annotation.HBaseColumn;
import com.yc.bigdata.hbase.annotation.HBaseTable;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.BufferedMutator;
import org.apache.hadoop.hbase.client.BufferedMutatorParams;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.client.TableDescriptor;
import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.util.StopWatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;

import java.io.*;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import cn.hutool.core.exceptions.ValidateException;

/**
 * <p></p>
 *
 * @author: YuanChilde
 * @date: 2020-01-16 14:03
 * @version: 1.0
 * Modification History:
 * Date    Author      Version     Description
 * -----------------------------------------------------------------
 * 2020-01-16 14:03    YuanChilde     1.0        新增
 */
public class HBaseTemplate implements HBaseOperations {

    private static final Logger LOGGER = LoggerFactory.getLogger(HBaseTemplate.class);

    private Configuration configuration;

    private volatile Connection connection;

    private Admin admin;

    public HBaseTemplate(Configuration configuration) {
        this.setConfiguration(configuration);
        try {
            this.admin = getConnection().getAdmin();
        } catch (IOException e) {
            LOGGER.error("HBaseTemplate实例初始化失败！错误信息为：" + e.getMessage(), e);
        }
    }

    @Override
    public <T> T execute(String tableName, TableCallback<T> action) {
        Assert.notNull(action, "TableCallback object must not be null");
        Assert.notNull(tableName, "No table specified");

        StopWatch sw = new StopWatch();
        sw.start();
        Table table = null;
        try {
            table = this.getTable(tableName);
            return action.doInTable(table);
        } catch (Throwable throwable) {
            throw new HBaseSystemException(throwable);
        } finally {
            if (null != table) {
                try {
                    table.close();
                    sw.stop();
                } catch (IOException e) {
                    LOGGER.error("hbase资源释放失败");
                }
            }
        }
    }

    @Override
    public <T> List<T> find(String tableName, String family, final RowMapper<T> action) {
        Scan scan = new Scan();
        scan.setCaching(5000);
        scan.addFamily(Bytes.toBytes(family));
        return this.find(tableName, scan, action);
    }

    @Override
    public <T> List<T> find(String tableName, String family, String qualifier, final RowMapper<T> action) {
        Scan scan = new Scan();
        scan.setCaching(5000);
        scan.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
        return this.find(tableName, scan, action);
    }

    @Override
    public <T> List<T> find(String tableName, final Scan scan, final RowMapper<T> action) {
        return this.execute(tableName, table -> {
            int caching = scan.getCaching();
            // 如果caching未设置(默认是1)，将默认配置成5000
            if (caching == 1) {
                scan.setCaching(5000);
            }
            ResultScanner scanner = table.getScanner(scan);
            try {
                List<T> rs = new ArrayList<T>();
                int rowNum = 0;
                for (Result result : scanner) {
                    rs.add(action.mapRow(result, rowNum++));
                }
                return rs;
            } finally {
                scanner.close();
            }
        });
    }

    @Override
    public <T> T get(String tableName, String rowName, final RowMapper<T> mapper) {
        return this.get(tableName, rowName, null, null, mapper);
    }

    @Override
    public <T> T get(String tableName, String rowName, String familyName, final RowMapper<T> mapper) {
        return this.get(tableName, rowName, familyName, null, mapper);
    }

    @Override
    public <T> T get(String tableName, final String rowName, final String familyName, final String qualifier, final RowMapper<T> mapper) {
        return this.execute(tableName, table -> {
            Get get = new Get(Bytes.toBytes(rowName));
            if (StringUtils.isNotBlank(familyName)) {
                byte[] family = Bytes.toBytes(familyName);
                if (StringUtils.isNotBlank(qualifier)) {
                    get.addColumn(family, Bytes.toBytes(qualifier));
                }
                else {
                    get.addFamily(family);
                }
            }
            Result result = table.get(get);
            return mapper.mapRow(result, 0);
        });
    }

    @Override
    public void execute(String tableName, MutatorCallback action) {
        Assert.notNull(action, "MutatorCallback object must not be null");
        Assert.notNull(tableName, "No table specified");

        StopWatch sw = new StopWatch();
        sw.start();
        BufferedMutator mutator = null;
        try {
            BufferedMutatorParams mutatorParams = new BufferedMutatorParams(TableName.valueOf(tableName));
            mutator = this.getConnection().getBufferedMutator(mutatorParams.writeBufferSize(3 * 1024 * 1024));
            action.doInMutator(mutator);
        } catch (Throwable throwable) {
            sw.stop();
            throw new HBaseSystemException(throwable);
        } finally {
            if (null != mutator) {
                try {
                    mutator.flush();
                    mutator.close();
                    sw.stop();
                } catch (IOException e) {
                    LOGGER.error("hbase mutator资源释放失败");
                }
            }
        }
    }

    @Override
    public void saveOrUpdate(String tableName, final Mutation mutation) {
        this.execute(tableName, mutator -> {
            mutator.mutate(mutation);
        });
    }

    @Override
    public void saveOrUpdates(String tableName, final List<Mutation> mutations) {
        this.execute(tableName, mutator -> {
            mutator.mutate(mutations);
        });
    }

    @Override
    public void putData(String tableName, String rowKey, String columnFamily, String column, String value) {
        if (StringUtils.isBlank(column)) {
            throw new ValidateException(MessageFormat.format("列名数据不能为空,tableName:{0},rowKey:{1},familyName:{2}", tableName, rowKey,
                    columnFamily));
        }

        Table table = null;
        try {
            table = this.getTable(tableName);
            Put put = new Put(Bytes.toBytes(rowKey));
            put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));
            table.put(put);
            if(LOGGER.isDebugEnabled()){
                LOGGER.debug("putData add or update data Success,rowKey:" + rowKey);
            }
        } catch (Exception e) {
            throw new HBaseSystemException(MessageFormat.format("为表添加 or 更新数据失败,tableName:{0},rowKey:{1},familyName:{2}", tableName, rowKey, columnFamily), e);
        } finally {
            close(null, null, table);
        }
    }

    @Override
    public void putDatas(String tableName,  String rowKey, String columnFamily, String[] columns, String[] values) {
        if (columns != null && values != null && columns.length == values.length) {
            for (int i = 0; i < columns.length; i++) {
                this.putData(tableName, rowKey, columnFamily, columns[i], values[i]);
            }
        } else {
            throw new ValidateException("数据为空或者列和值数量不一致");
        }
    }

    @Override
    public Map<String, String> selectRow(String tableName, String rowKey) {
        Get get = new Get(rowKey.getBytes());
        Table table = null;
        try {
            Map<String, String> result = Maps.newHashMap();
            table = getTable(tableName);
            Result hTableResult = table.get(get);
            if (hTableResult != null && !hTableResult.isEmpty()) {
                for (Cell cell : hTableResult.listCells()) {
                    result.put(
                            Bytes.toString(cell.getQualifierArray(),
                                    cell.getQualifierOffset(),
                                    cell.getQualifierLength()),
                            Bytes.toString(cell.getValueArray(),
                                    cell.getValueOffset(),
                                    cell.getValueLength()));
                }
            }
            return result;
        } catch (IOException e) {
            throw new HBaseSystemException(MessageFormat.format(
                    "查询一行的数据失败,tableName:{0},rowKey:{1}", tableName, rowKey), e);
        } finally {
            close(null, null, table);
        }
    }

    @Override
    public String selectValue(String tableName, String rowKey, String columnFamily, String column) {
        Get get = new Get(rowKey.getBytes());
        Table table = null;
        try {
            table = getTable(tableName);
            get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column));
            Result rs = table.get(get);
            return Bytes.toString(rs.value());
        } catch (IOException e) {
            throw new HBaseSystemException(MessageFormat.format("查询一行的数据失败,tableName:{0},rowKey:{1}", tableName, rowKey), e);
        } finally {
            close(null, null, table);
        }
    }

    @Override
    public Result getRow(String tableName, String rowKey, FilterList filterList) {
        try (Table table = getTable(tableName)) {
            Get get = new Get(Bytes.toBytes(rowKey));
            get.setFilter(filterList);
            return table.get(get);
        } catch (IOException e) {
            throw new HBaseSystemException(MessageFormat.format("查询数据失败,tableName:{0},rowKey:{1}", tableName, rowKey), e);
        }
    }

    @Override
    public ResultScanner getScanner(String tableName, String startRowKey, String endRowKey) {
        return this.getScanner(tableName, startRowKey, endRowKey, null);
    }

    @Override
    public ResultScanner getScanner(String tableName, String startRowKey, String endRowKey, FilterList filterList) {
        try (Table table = getTable(tableName)) {
            Scan scan = new Scan();
            scan.withStartRow(Bytes.toBytes(startRowKey));
            scan.withStopRow(Bytes.toBytes(endRowKey));
            if (filterList != null) {
                scan.setFilter(filterList);
            }
            scan.setCaching(1000);
            return table.getScanner(scan);
        } catch (IOException e) {
            throw new HBaseSystemException(MessageFormat.format("查询数据失败,tableName:{0},startRowKey:{1},endRowKey:{2}", tableName,
                    startRowKey, endRowKey), e);
        }
    }

    @Override
    public Map<String, Map<String, String>> scanAllRecord(String tableName) throws IOException {
        Map<String, Map<String, String>> result = Maps.newHashMap();
        Table table = null;
        ResultScanner rs = null;
        try {
            table = getTable(tableName);
            Scan scan = new Scan();
            rs = table.getScanner(scan);
            for (Result r : rs) {
                // 每一行数据
                Map<String, String> columnMap = Maps.newHashMap();
                String rowKey = null;
                for (Cell cell : r.listCells()) {
                    if (rowKey == null) {
                        rowKey = Bytes.toString(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
                    }
                    columnMap.put(Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()),
                            Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
                }
                if (rowKey != null) {
                    result.put(rowKey, columnMap);
                }
            }

        } finally {
            close(null, rs, table);
        }
        return result;
    }

    @Override
    public void createTable(String tableName, List<String> columnFamily) throws IOException {
        TableName name = TableName.valueOf(tableName);
        if (this.admin.tableExists(name)) {
           throw new ValidateException("该表已存在");
        }

        List<ColumnFamilyDescriptor> cfDesc = new ArrayList<>(columnFamily.size());
        columnFamily.forEach(cf -> {
            cfDesc.add(ColumnFamilyDescriptorBuilder.newBuilder(
                    Bytes.toBytes(cf)).build());
        });

        TableDescriptor tableDesc = TableDescriptorBuilder
                .newBuilder(name)
                .setColumnFamilies(cfDesc).build();
        this.admin.createTable(tableDesc);
    }

    @Override
    public void createOrOverwriteTable(String tableName, List<String> columnFamily) throws IOException {
        this.deleteTable(tableName);
        this.createTable(tableName, columnFamily);
    }

    @Override
    public void addColumnFamily(String tableName, String columnFamily) throws IOException {
        // 新增一个列族
        ColumnFamilyDescriptorBuilder.ModifyableColumnFamilyDescriptor columnFamilyDescriptorBuilder =
                (ColumnFamilyDescriptorBuilder.ModifyableColumnFamilyDescriptor) ColumnFamilyDescriptorBuilder.of(columnFamily);
        columnFamilyDescriptorBuilder.setCompactionCompressionType(Compression.Algorithm.GZ);
        columnFamilyDescriptorBuilder.setMaxVersions(HConstants.ALL_VERSIONS);
        // 执行新增操作
        admin.addColumnFamily(TableName.valueOf(tableName), columnFamilyDescriptorBuilder);
    }

    @Override
    public List<String> getAllTableNames() throws IOException {
        List<String> result = new ArrayList<>();
        TableName[] tableNames = admin.listTableNames();
        for (TableName tableName : tableNames) {
            result.add(tableName.getNameAsString());
        }
        return result;
    }

    @Override
    public void deleteRow(String tablename, String rowkey) throws IOException {
        Table table = this.getTable(tablename);
        Delete d = new Delete(rowkey.getBytes());
        table.delete(d);
        close(null, null, table);
    }

    @Override
    public void deleteColumnFamily(String tablename, String rowkey, String columnFamily) throws IOException {
        Table table = this.getTable(tablename);
        Delete d = new Delete(rowkey.getBytes()).addFamily(Bytes.toBytes(columnFamily));
        table.delete(d);
    }

    @Override
    public void deleteColumn(String tablename, String rowkey, String columnFamily, String column) throws IOException {
        Table table = this.getTable(tablename);
        Delete d = new Delete(rowkey.getBytes()).addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column));
        table.delete(d);
    }


    @Override
    public void deleteTable(String tableName) throws IOException {
        TableName name = TableName.valueOf(tableName);
        if (admin.tableExists(name)) {
            admin.disableTable(name);
            admin.deleteTable(name);
        }
    }

    public Connection getConnection() {
        if (null == this.connection) {
            synchronized (this) {
                if (null == this.connection) {
                    try {
                        ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(5, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
                        poolExecutor.prestartCoreThread();
                        this.connection = ConnectionFactory.createConnection(configuration, poolExecutor);
                    } catch (IOException e) {
                        LOGGER.error("hbase connection资源池创建失败");
                    }
                }
            }
        }
        return this.connection;
    }

    /**
     * JavaBean转换为Put
     * @param <T>
     * @param obj
     * @return
     * @throws Exception
     */
    public <T> Put beanToPut(T obj) throws Exception {
        Put put = new Put(Bytes.toBytes(parseObjId(obj)));
        Class<?> clazz = obj.getClass();
        Field[] fields = clazz.getDeclaredFields();
        for (Field field : fields) {
            if (!field.isAnnotationPresent(HBaseColumn.class)) {
                continue;
            }
            field.setAccessible(true);
            HBaseColumn orm = field.getAnnotation(HBaseColumn.class);
            String family = orm.family();
            String qualifier = orm.qualifier();
            if (org.apache.commons.lang.StringUtils.isBlank(family) || org.apache.commons.lang.StringUtils.isBlank(qualifier)) {
                continue;
            }
            Object fieldObj = field.get(obj);
            if (fieldObj.getClass().isArray()) {
                LOGGER.error("nonsupport");
            }
            if ("rowkey".equalsIgnoreCase(qualifier) || "rowkey".equalsIgnoreCase(family)) {
                continue;
            }
            if (field.get(obj) != null || org.apache.commons.lang.StringUtils.isNotBlank(field.get(obj).toString())) {
                put.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes.toBytes(field.get(obj).toString()));
            }
        }
        return put;
    }

    /**
     * 获取Bean中的id,作为Rowkey
     * @param <T>
     *
     * @param obj
     * @return
     */
    public <T> String parseObjId(T obj) {
        Class<?> clazz = obj.getClass();
        try {
            Field field = clazz.getDeclaredField("id");
            field.setAccessible(true);
            Object object = field.get(obj);
            return object.toString();
        } catch (NoSuchFieldException e) {
            LOGGER.error("", e);
        } catch (SecurityException e) {
            LOGGER.error("", e);
        } catch (IllegalArgumentException e) {
            LOGGER.error("", e);
        } catch (IllegalAccessException e) {
            LOGGER.error("", e);
        }
        return "";
    }

    /**
     * HBase result 转换为 bean
     * @param <T>
     * @param result
     * @param obj
     * @return
     * @throws Exception
     */
    public <T> T resultToBean(Result result, T obj) throws Exception {
        if (result == null) {
            return null;
        }
        Class<?> clazz = obj.getClass();
        Field[] fields = clazz.getDeclaredFields();
        for (Field field : fields) {
            if (!field.isAnnotationPresent(HBaseColumn.class)) {
                continue;
            }
            HBaseColumn orm = field.getAnnotation(HBaseColumn.class);
            String family = orm.family();
            String qualifier = orm.qualifier();
            boolean timeStamp = orm.timestamp();
            if (org.apache.commons.lang.StringUtils.isBlank(family) || org.apache.commons.lang.StringUtils.isBlank(qualifier)) {
                continue;
            }
            String fieldName = field.getName();
            String value = "";
            if ("rowkey".equalsIgnoreCase(family)) {
                value = new String(result.getRow());
            } else {
                value = getResultValueByType(result, family, qualifier, timeStamp);
            }
            String firstLetter = fieldName.substring(0, 1).toUpperCase();
            String setMethodName = "set" + firstLetter + fieldName.substring(1);
            Method setMethod = clazz.getMethod(setMethodName, new Class[] { field.getType() });
            setMethod.invoke(obj, new Object[] { value });
        }
        return obj;
    }

    /**
     * @param result
     * @param family
     * @param qualifier
     * @param timeStamp
     * @return
     */
    private String getResultValueByType(Result result, String family, String qualifier, boolean timeStamp) {
        if (!timeStamp) {
            return new String(result.getValue(Bytes.toBytes(family), Bytes.toBytes(qualifier)));
        }
        List<Cell> cells = result.getColumnCells(Bytes.toBytes(family), Bytes.toBytes(qualifier));
        if (cells.size() == 1) {
            Cell cell = cells.get(0);
            return cell.getTimestamp() + "";
        }
        return "";
    }

    /**
     * 从对象注解中获取表名
     * @param obj
     * @return
     */
    private String getORMTable(Object obj) {
        HBaseTable table = obj.getClass().getAnnotation(HBaseTable.class);
        return table.tableName();
    }

    /**
     * @Descripton: 根据条件过滤查询
     * @Author: Sorin
     * @param obj
     * @param param
     * @Date: 2018/3/26
     */
    public <T> List<T> queryScan(T obj, Map<String, String> param)throws Exception{
        List<T> objs = new ArrayList<T>();
        String tableName = getORMTable(obj);
        if (org.apache.commons.lang.StringUtils.isBlank(tableName)) {
            return null;
        }
        try (Table table = this.getTable(tableName);Admin admin = this.getAdmin();){
            if(!admin.isTableAvailable(TableName.valueOf(tableName))){
                return objs;
            }
            Scan scan = new Scan();
            for (Map.Entry<String, String> entry : param.entrySet()){
                Class<?> clazz = obj.getClass();
                Field[] fields = clazz.getDeclaredFields();
                for (Field field : fields) {
                    if (!field.isAnnotationPresent(HBaseColumn.class)) {
                        continue;
                    }
                    field.setAccessible(true);
                    HBaseColumn orm = field.getAnnotation(HBaseColumn.class);
                    String family = orm.family();
                    String qualifier = orm.qualifier();
                    if(qualifier.equals(entry.getKey())){
                        Filter filter = new SingleColumnValueFilter(Bytes.toBytes(family), Bytes.toBytes(entry.getKey()), CompareOperator.EQUAL, Bytes.toBytes(entry.getValue()));
                        scan.setFilter(filter);
                    }
                }
            }
            ResultScanner scanner = table.getScanner(scan);
            for (Result result : scanner) {
                T beanClone = (T)BeanUtils.cloneBean(this.resultToBean(result, obj));
                objs.add(beanClone);
            }
        } catch (Exception e) {
            e.printStackTrace();
            LOGGER.error("查询失败！");
            throw new Exception(e);
        }
        return objs;
    }

    /**
     * @Descripton: 根据rowkey查询
     * @Author: Sorin
     * @param obj
     * @param rowkeys
     * @Date: 2018/3/22
     */
    public <T> List<T> get(T obj, String ... rowkeys) {
        List<T> objs = new ArrayList<T>();
        String tableName = getORMTable(obj);
        if (org.apache.commons.lang.StringUtils.isBlank(tableName)) {
            return objs;
        }
        try (Table table = this.getTable(tableName);Admin admin = this.getAdmin()){
            if(!admin.isTableAvailable(TableName.valueOf(tableName))){
                return objs;
            }
            List<Result> results = getResults(tableName, rowkeys);
            if (results.isEmpty()) {
                return objs;
            }
            for (int i = 0; i < results.size(); i++) {
                T bean = null;
                Result result = results.get(i);
                if (result == null || result.isEmpty()) {
                    continue;
                }
                try {
                    bean = this.resultToBean(result, obj);
                    objs.add(bean);
                } catch (Exception e) {
                    e.printStackTrace();
                    LOGGER.error("查询异常！", e);
                }
            }
        }catch (Exception e){
            e.printStackTrace();
        }
        return objs;
    }


    /**
     * @Descripton: 保存实体对象
     * @Author: Sorin
     * @param objs
     * @Date: 2018/3/22
     */
    public <T> boolean save(T ... objs) {
        List<Put> puts = new ArrayList<Put>();
        String tableName = "";
        try (Admin admin = this.getAdmin()){
            for (Object obj : objs) {
                if (obj == null) {
                    continue;
                }
                tableName = getORMTable(obj);
                // 表不存在，先获取family创建表
                if(!admin.isTableAvailable(TableName.valueOf(tableName))){
                    // 获取family, 创建表
                    Class<?> clazz = obj.getClass();
                    Field[] fields = clazz.getDeclaredFields();
                    List<String> set = new ArrayList<>(10);
                    for(int i=0;i<fields.length;i++){
                        if (!fields[i].isAnnotationPresent(HBaseColumn.class)) {
                            continue;
                        }
                        fields[i].setAccessible(true);
                        HBaseColumn orm = fields[i].getAnnotation(HBaseColumn.class);
                        String family = orm.family();
                        if ("rowkey".equalsIgnoreCase(family)) {
                            continue;
                        }
                        set.add(family);
                    }
                    // 创建表
                    createTable(tableName, set);
                }
                Put put = this.beanToPut(obj);
                puts.add(put);
            }
        }catch (Exception e){
            e.printStackTrace();
            LOGGER.error("保存Hbase异常！");
        }
        return savePut(puts, tableName);
    }

    /**
     * @Descripton: 根据tableName保存
     * @Author: Sorin
     * @param tableName
     * @param objs
     * @Date: 2018/3/22
     */
    public <T> void save(String tableName, T ... objs){
        List<Put> puts = new ArrayList<Put>();
        for (Object obj : objs) {
            if (obj == null) {
                continue;
            }
            try {
                Put put = this.beanToPut(obj);
                puts.add(put);
            } catch (Exception e) {
                LOGGER.warn("", e);
            }
        }
        savePut(puts, tableName);
    }

    /**
     * @Descripton: 删除
     * @Author: Sorin
     * @param obj
     * @param rowkeys
     * @Date: 2018/3/22
     */
    public <T> void delete(T obj, String... rowkeys) {
        String tableName = "";
        tableName = getORMTable(obj);
        if (org.apache.commons.lang.StringUtils.isBlank(tableName)) {
            return;
        }
        List<Delete> deletes = new ArrayList<Delete>();
        for (String rowkey : rowkeys) {
            if (org.apache.commons.lang.StringUtils.isBlank(rowkey)) {
                continue;
            }
            deletes.add(new Delete(Bytes.toBytes(rowkey)));
        }
        delete(deletes, tableName);
    }


    /**
     * @Descripton: 批量删除
     * @Author: Sorin
     * @param deletes
     * @param tableName
     * @Date: 2018/3/22
     */
    private void delete(List<Delete> deletes, String tableName) {
        try (Table table = this.getTable(tableName)) {
            if (org.apache.commons.lang.StringUtils.isBlank(tableName)) {
                LOGGER.info("tableName为空！");
                return;
            }
            table.delete(deletes);
        } catch (IOException e) {
            e.printStackTrace();
            LOGGER.error("删除失败！",e);
        }
    }

    /**
     * @Descripton: 根据tableName获取列簇名称
     * @Author: Sorin
     * @param tableName
     * @Date: 2018/3/22
     */
    public List<String> familys(String tableName) {
        try (Table table = this.getTable(tableName)){
            List<String> columns = new ArrayList<>();
            if (table==null) {
                return columns;
            }
            HTableDescriptor tableDescriptor = table.getTableDescriptor();
            HColumnDescriptor[] columnDescriptors = tableDescriptor.getColumnFamilies();
            for (HColumnDescriptor columnDescriptor :columnDescriptors) {
                String columnName = columnDescriptor.getNameAsString();
                columns.add(columnName);
            }
            return columns;
        } catch (Exception e) {
            e.printStackTrace();
            LOGGER.error("查询列簇名称失败！" ,e);
        }
        return new ArrayList<String>();
    }

    // 保存方法
    private boolean savePut(List<Put> puts, String tableName) {
        if (org.apache.commons.lang.StringUtils.isBlank(tableName)) {
            return false;
        }
        try (Table table = this.getTable(tableName); Admin admin = this.getAdmin()) {
            table.put(puts);
            return true;
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
    }

    // 获取查询结果
    private List<Result> getResults(String tableName, String... rowkeys) {
        List<Result> resultList = new ArrayList<>();
        List<Get> gets = new ArrayList<>();
        for (String rowkey : rowkeys) {
            if (org.apache.commons.lang.StringUtils.isBlank(rowkey)) {
                continue;
            }
            Get get = new Get(Bytes.toBytes(rowkey));
            gets.add(get);
        }
        try (Table table = this.getTable(tableName)) {
            Result[] results = table.get(gets);
            Collections.addAll(resultList, results);
            return resultList;
        } catch (Exception e) {
            e.printStackTrace();
            return resultList;
        }
    }

    /**
     * @Descripton: 根据条件过滤查询（大于等于）
     * @Author: Sorin
     * @param obj
     * @param param
     * @Date: 2018/3/26
     */
    public <T> List<T> queryScanGreater(T obj, Map<String, String> param)throws Exception{
        List<T> objs = new ArrayList<T>();
        String tableName = getORMTable(obj);
        if (org.apache.commons.lang.StringUtils.isBlank(tableName)) {
            return null;
        }
        try (Table table = this.getTable(tableName);Admin admin = this.getAdmin()){
            if(!admin.isTableAvailable(TableName.valueOf(tableName))){
                return objs;
            }
            Scan scan = new Scan();
            for (Map.Entry<String, String> entry : param.entrySet()){
                Class<?> clazz = obj.getClass();
                Field[] fields = clazz.getDeclaredFields();
                for (Field field : fields) {
                    if (!field.isAnnotationPresent(HBaseColumn.class)) {
                        continue;
                    }
                    field.setAccessible(true);
                    HBaseColumn orm = field.getAnnotation(HBaseColumn.class);
                    String family = orm.family();
                    String qualifier = orm.qualifier();
                    if(qualifier.equals(entry.getKey())){
                        Filter filter = new SingleColumnValueFilter(Bytes.toBytes(family), Bytes.toBytes(entry.getKey()), CompareOperator.GREATER_OR_EQUAL, Bytes.toBytes(entry.getValue()));
                        scan.setFilter(filter);
                    }
                }
            }
            ResultScanner scanner = table.getScanner(scan);
            for (Result result : scanner) {
                T beanClone = (T)BeanUtils.cloneBean(this.resultToBean(result, obj));
                objs.add(beanClone);
            }
        } catch (Exception e) {
            e.printStackTrace();
            LOGGER.error("查询失败！");
            throw new Exception(e);
        }
        return objs;
    }

    /**
     * 根据rowkey查询记录
     * @param obj
     * @param rowkey
     * @param <T>
     * @return
     */
    public <T> List<T> queryScanRowkey(T obj, String rowkey){
        List<T> objs = new ArrayList<T>();
        String tableName = getORMTable(obj);
        if (org.apache.commons.lang.StringUtils.isBlank(tableName)) {
            return null;
        }
        ResultScanner scanner = null;
        try (Table table = this.getTable(tableName);Admin admin = this.getAdmin()){
            Scan scan = new Scan();
            scan.setRowPrefixFilter(Bytes.toBytes(rowkey));
            scanner = table.getScanner(scan);
            for (Result result : scanner) {
                T beanClone = (T) BeanUtils.cloneBean(this.resultToBean(result, obj));
                objs.add(beanClone);
            }
        }catch (Exception e){
            LOGGER.error("queryScanRowkey:查询失败！", e);
        }finally {
            if(scanner!=null){
                try {
                    scanner.close();
                } catch (Exception e) {
                    LOGGER.error("queryScan:关闭流异常！", e);
                }
            }
        }
        return objs;
    }

    /**
     * 根据表名 获取table
     */
    public Table getTable(String tableName) throws IOException {
        return connection.getTable(TableName.valueOf(tableName));
    }

    /**
     * 关闭流
     */
    public void close(Admin admin, ResultScanner rs, Table table) {
        if (admin != null) {
            try {
                admin.close();
            } catch (IOException e) {
                LOGGER.error("关闭Admin失败", e);
            }
        }

        if (rs != null) {
            rs.close();
        }

        if (table != null) {
            try {
                table.close();
            } catch (IOException e) {
                LOGGER.error("关闭Table失败", e);
            }
        }
    }

    public Configuration getConfiguration() {
        return configuration;
    }

    public Admin getAdmin() {
        return admin;
    }

    public void setConfiguration(Configuration configuration) {
        Assert.notNull(configuration, " a valid configuration is required");
        this.configuration = configuration;
    }

}
